package com.cnpc.bds.sdk.sink.kafka;

import com.cnpc.bds.sdk.sink.kafka.executor.DefaultUpstreamConsumer;
import com.cnpc.bds.sdk.sink.kafka.executor.UpstreamConsumer;
import com.cnpc.bds.sdk.sink.kafka.meta.ConsumerCfg;
import com.cnpc.bds.sdk.sink.kafka.meta.TopicMeta;

import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

/**
 * 上游消费任务管理器.
 */
public class UpstreamManager {

    private static final Map<TopicMeta, UpstreamConsumer> UPSTREAM_CONSUMER_TABLE = new ConcurrentHashMap<>();

    public static UpstreamConsumer allocate(ConsumerCfg consumerCfg) {
        return UPSTREAM_CONSUMER_TABLE.computeIfAbsent(consumerCfg.getTopicMeta(), meta -> DefaultUpstreamConsumer.buildAndExec(consumerCfg));
    }

}
